package com.example.service.pulsar;

import com.example.config.PulsarConfiguration;
import com.example.ececption.PulsarAutoConfigException;
import com.example.listener.BaseMessageListener;
import com.example.utils.TopicUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.shade.org.apache.commons.lang.StringUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author yuanzheng
 * @date 2020/7/26-18:13
 */
@Slf4j
@Configuration
@ConditionalOnWebApplication
@EnableConfigurationProperties(PulsarConfiguration.class)
public class PulsarServiceAutoConfiguration {

    @Resource
    private PulsarConfiguration pulsarConfiguration;

    @Bean
    public PulsarService messageService() {
        // 1.实例化客户端
        Map<String, PulsarClient> clients = new HashMap<>(8);
        List<PulsarConfiguration.PulsarConfig> clientList = pulsarConfiguration.clients;
        clientList.forEach(config -> {
            String serviceUrl = config.getServiceUrl();
            if (StringUtils.isNotBlank(serviceUrl)) {
                try {
                    PulsarClient client = PulsarClient.builder().serviceUrl(serviceUrl)
                            .enableTcpNoDelay(config.getEnableTcpNoDelay())
                            .operationTimeout(config.getOperationTimeout(), TimeUnit.SECONDS)
                            .listenerThreads(config.getListenerThreads())
                            .ioThreads(config.getIoThreads())
                            .build();
                    String id = config.getId();
                    log.info("Pulsar Client 实例化成功, id is {}, serviceUrl is {}", id, serviceUrl);
                    clients.put(id, client);
                } catch (PulsarClientException e) {
                    log.error("Pulsar Client 实例化失败！");
                    throw new PulsarAutoConfigException("Pulsar Client实例化失败！", e);
                }
            }
        });
        // 2.实例化监听
        clientList.forEach(config -> {
            try {
                // 是否配置对主题的监听
                String topicNames = config.getTopicNames();
                String[] topicNameArr = topicNames.split(",");
                String subscriptionName = config.getSubscriptionName();
                int length = topicNameArr.length;
                if (length > 0) {
                    // 验证是否初始化成功
                    String id = config.getId();
                    PulsarClient client = clients.get(id);
                    if (Objects.isNull(client)) {
                        log.error("Pulsar 数据源对应 PulsarClient 不存在 id is {}", id);
                    }
                    // 初始化监听
                    BaseMessageListener baseMessageListener = new BaseMessageListener();
                    ConsumerBuilder<String> consumerBuilder = client.newConsumer(Schema.STRING).receiverQueueSize(config.getReceiverQueueSize());
                    if (Boolean.TRUE.equals(baseMessageListener.enableAsync())) {
                        log.info("Pulsar 消费者开启异步消费 开始初始化消费线程池....");
                        baseMessageListener.initThreadPool(config.getCoreThreads(), config.getMaxCoreThreads(),
                                config.getKeepAliveTime(), config.getMaxQueueLength(), id);
                    }
                    String tenant = config.getTenant();
                    String namespace = config.getNamespace();
                    // 监听主题
                    List<String> topics = new ArrayList<>(length);
                    // 处理Topic
                    Boolean persistent = config.isPersistent();
                    for (String topic : topicNameArr) {
                        topics.add(TopicUtil.generateTopic(persistent, tenant, namespace, topic));
                    }
                    consumerBuilder.topics(topics);
                    // 处理订阅名称
                    consumerBuilder.subscriptionName(subscriptionName);
                    consumerBuilder.ackTimeout(config.getAckTimeout(), TimeUnit.MILLISECONDS);
                    consumerBuilder.subscriptionType(config.getSubscriptionType());
                    // 处理死信策略
                    boolean enableRetry = config.isEnableRetry();
                    if (Boolean.TRUE.equals(enableRetry)) {
                        DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
                                .maxRedeliverCount(config.getMaxRedeliverCount())
                                .build();
                        if (StringUtils.isNotBlank(config.getRetryLetterTopic())) {
                            deadLetterPolicy.setRetryLetterTopic(config.getRetryLetterTopic());
                        }
                        if (StringUtils.isNotBlank(config.getDeadLetterTopic())) {
                            deadLetterPolicy.setDeadLetterTopic(config.getDeadLetterTopic());
                        }
                        consumerBuilder.enableRetry(true).deadLetterPolicy(deadLetterPolicy);
                    } else {
                        if (StringUtils.isNotBlank(config.getDeadLetterTopic())) {
                            if (SubscriptionType.Exclusive.equals(config.getSubscriptionType())) {
                                // throw new PulsarAutoConfigException("Pulsar 消费端仅支持在Shared/Key_Shared模式下单独使用死信队列");
                                log.error("Pulsar 消费端仅支持在Shared/Key_Shared模式下单独使用死信队列");
                            } else {
                                DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
                                        .maxRedeliverCount(config.getMaxRedeliverCount())
                                        .deadLetterTopic(config.getRetryLetterTopic())
                                        .build();
                                consumerBuilder.deadLetterPolicy(deadLetterPolicy);
                            }
                        }
                    }
                    consumerBuilder.messageListener(baseMessageListener);
                    Consumer<String> consumer = consumerBuilder.subscribe();
                    log.info("Pulsar Consumer初始化完毕, id is {}, topic is {},", id, consumer.getTopic());
                }
            } catch (PulsarClientException e) {
                throw new PulsarAutoConfigException("[Pulsar] consumer初始化异常", e);
            }
        });
        // 实例化service
        PulsarService service = new PulsarService();
        service.setClients(clients);
        return service;
    }
}
